﻿using Microsoft.Extensions.Logging;
using StackExchange.Redis;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Channels;
using System.Threading.Tasks;
using TouchSocket.Sockets;
using YS.Domain.IoT.Const;
using YS.Domain.IoT.Extension;
using YS.Domain.IoT.Option;

namespace YS.K12.Application.IoT.RedisHub
{
    /// <summary>
    /// Redis Socket集群
    /// </summary>
    public class RedisHubLifetimeManager
    {
        private readonly ILogger<RedisHubLifetimeManager> _logger;
        private IConnectionMultiplexer _redisServerConnection;
        private ISubscriber _bus;
        private readonly ITcpService _tcpService;
        private readonly SemaphoreSlim _connectionLock = new SemaphoreSlim(1);
        /// <summary>
        /// 避免重复登录导致的重复订阅
        /// </summary>
        public  readonly ConcurrentDictionary<string,bool> _subscriptions = new ConcurrentDictionary<string, bool>();
        /// <summary>
        ///
        /// </summary>
        /// <param name="logger"></param>
        /// <param name="tcpService"></param>
        public RedisHubLifetimeManager(ILogger<RedisHubLifetimeManager> logger, ITcpService tcpService)
        {
            _logger = logger;
            _tcpService = tcpService;
            _ = EnsureRedisServerConnection();
        }

        public async Task SubscribeToConnection(ITcpSession client)
        {
            var mid = client.GetClientId();
            var connectionChannel = CacheConst.ConnectionChannelKey.ToFormat(mid);
            // 检查是否已经订阅
            if (!_subscriptions.TryAdd(connectionChannel, true))
            {
                //_logger.LogInformation("客户端 {Mid} 已经订阅了通道 {Channel}", mid, connectionChannel);
                return;
            }
            
            var channel = await _bus!.SubscribeAsync(connectionChannel);
            channel.OnMessage(async channelMessage =>
            {
                _logger.LogInformation("订阅消息:{Mid}", mid);
                await _tcpService.SendAsync(mid, Encoding.UTF8.GetBytes(channelMessage.Message).AsMemory<byte>());
            });
        }

        public async Task PublishAsync(string clientId, ReadOnlyMemory<byte> payload)
        {
            await EnsureRedisServerConnection();
            var channel = CacheConst.ConnectionChannelKey.ToFormat(clientId);
            await _bus!.PublishAsync(channel, payload);
        }
        /// <summary>
        ///删除订阅
        /// </summary>
        /// <param name="clientId"></param>
        /// <returns></returns>
        public async Task UnsubscribeFromConnection(string clientId)
        {
            var connectionChannel = CacheConst.ConnectionChannelKey.ToFormat(clientId);
            //// 从 Redis 中移除订阅状态
            //await RemoveSubscriptionAsync(connectionChannel);
            _subscriptions.TryRemove(connectionChannel, out _);
            // 取消订阅
            await _bus!.UnsubscribeAsync(connectionChannel);
        }
        public async Task EnsureRedisServerConnection()
        {
            if (_redisServerConnection != null)
            {
                return;
            }
            try
            {
                await _connectionLock.WaitAsync();
                if (_redisServerConnection != null)
                {
                    return;
                } 
                 _redisServerConnection = ConnectionMultiplexer.Connect(AppSettingsHelper.GetContent("Iot-Redis-Main"));
   

                _bus = _redisServerConnection.GetSubscriber();
            }
            catch (Exception ex)
            {
                _logger.LogError($"EnsureRedisServerConnection:出错:{ex.Message}");
            }

        }
    }
}
